package com.gph.saviorframework.mq.service.impl;

import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
import com.alibaba.rocketmq.common.protocol.body.KVTable;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.alibaba.rocketmq.tools.command.broker.BrokerStatusSubCommand;
import com.alibaba.rocketmq.tools.command.cluster.ClusterListSubCommand;
import com.gph.saviorframework.mq.service.ClusterService;
import com.gph.saviorframework.mq.service.impl.validate.CmdTrace;

import java.util.*;

/**
 * Created by root on 16-7-18.
 */
public class ClusterServiceImpl extends AbstractService implements ClusterService {

    @CmdTrace(cmdClazz = ClusterListSubCommand.class)
    public ClusterInfo getClusterInfo() {
        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        try{
            defaultMQAdminExt.start();
            ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
            return clusterInfo;
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
        return null;
    }

    /**
     * 获取最终结果集合，提供给页面使用
     *
     * @return
     */
    @CmdTrace(cmdClazz = ClusterListSubCommand.class)
    public List<Map<String,Object>> getClusterInfoForMap() {

        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        try{
            defaultMQAdminExt.start();
            return doProcessData(defaultMQAdminExt);
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
        return null;
    }

    private  List<Map<String,Object>> doProcessData(DefaultMQAdminExt defaultMQAdminExt){
        List<Map<String,Object>> list = new ArrayList<Map<String, Object>>();
        try{
            ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();

            HashMap<String,Set<String>> clusterAddrTable= clusterInfo.getClusterAddrTable();

            for(String clauster : clusterAddrTable.keySet()){
                //添加cluster名称和broker集合
                Map<String,Object> map = new LinkedHashMap<String, Object>();

                List<Map<String,Object>> listBroker = new ArrayList<Map<String, Object>>();

                for(String brokerName : clusterAddrTable.get(clauster)){
                    //构建broker
                    Map<String,Object> brokerMap = new LinkedHashMap<String, Object>();

                    //构建broker详细信息map
                    BrokerData brokerData = clusterInfo.getBrokerAddrTable().get(brokerName);
                    if (brokerData != null) {
                        Set<Map.Entry<Long, String>> brokerAddrSet = brokerData.getBrokerAddrs().entrySet();
                        Iterator<Map.Entry<Long, String>> itAddr = brokerAddrSet.iterator();

                        List<Map<String,Object>> listBrokerInstance = new ArrayList<Map<String, Object>>();

                        while (itAddr.hasNext()){

                            double in = 0;
                            double out = 0;
                            String version = "";

                            long InTotalYest = 0;
                            long OutTotalYest = 0;
                            long InTotalToday = 0;
                            long OutTotalToday = 0;

                            Map.Entry<Long, String> next1 = itAddr.next();
                            KVTable kvTable = defaultMQAdminExt.fetchBrokerRuntimeStats(next1.getValue());

                            String putTps = kvTable.getTable().get("putTps");
                            String getTransferedTps = kvTable.getTable().get("getTransferedTps");
                            version = kvTable.getTable().get("brokerVersionDesc");
                            {
                                String[] tpss = putTps.split(" ");
                                if (tpss != null && tpss.length > 0) {
                                    in = Double.parseDouble(tpss[0]);
                                }
                            }

                            {
                                String[] tpss = getTransferedTps.split(" ");
                                if (tpss != null && tpss.length > 0) {
                                    out = Double.parseDouble(tpss[0]);
                                }
                            }

                            String msgPutTotalYesterdayMorning =
                                    kvTable.getTable().get("msgPutTotalYesterdayMorning");
                            String msgPutTotalTodayMorning =
                                    kvTable.getTable().get("msgPutTotalTodayMorning");
                            String msgPutTotalTodayNow = kvTable.getTable().get("msgPutTotalTodayNow");
                            String msgGetTotalYesterdayMorning =
                                    kvTable.getTable().get("msgGetTotalYesterdayMorning");
                            String msgGetTotalTodayMorning =
                                    kvTable.getTable().get("msgGetTotalTodayMorning");
                            String msgGetTotalTodayNow = kvTable.getTable().get("msgGetTotalTodayNow");

                            InTotalYest =
                                    Long.parseLong(msgPutTotalTodayMorning)
                                            - Long.parseLong(msgPutTotalYesterdayMorning);
                            OutTotalYest =
                                    Long.parseLong(msgGetTotalTodayMorning)
                                            - Long.parseLong(msgGetTotalYesterdayMorning);

                            InTotalToday =
                                    Long.parseLong(msgPutTotalTodayNow)
                                            - Long.parseLong(msgPutTotalTodayMorning);
                            OutTotalToday =
                                    Long.parseLong(msgGetTotalTodayNow)
                                            - Long.parseLong(msgGetTotalTodayMorning);

                            Map<String,Object> brokerInstanceMap = new HashMap<String, Object>();
                            brokerInstanceMap.put("BID",next1.getKey().longValue());
                            brokerInstanceMap.put("Addr",next1.getValue());
                            brokerInstanceMap.put("Version",version);
                            brokerInstanceMap.put("InTPS",in);
                            brokerInstanceMap.put("OutTPS",out);
                            brokerInstanceMap.put("InTotalYest",InTotalYest);
                            brokerInstanceMap.put("OutTotalYest",OutTotalYest);
                            brokerInstanceMap.put("InTotalToday",InTotalToday);
                            brokerInstanceMap.put("OutTotalToday",OutTotalToday);

                            listBrokerInstance.add(brokerInstanceMap);
                        }

                        brokerMap.put(brokerName, listBrokerInstance);

                        listBroker.add(brokerMap);
                    }
                    map.put(clauster,listBroker);
                }
                list.add(map);
            }
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        return list;
    }
}
